/**
 * 
 */
package jvs.peer.io;

import java.io.IOException;
import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.io.UnsupportedEncodingException;

import jvs.peer.misc.Base64;
import jvs.peer.util.Log;


/**
 * @author qiangli
 *
 */
public class MessageInputStream extends PipedInputStream {
	private Channel channel;
	private PipedOutputStream out;
	private int seqno = 0; //messages received
	private int counter = 0; //bytes received
	
	public MessageInputStream(Channel channel) throws IOException {
		this.channel = channel;
		this.out = new PipedOutputStream(this);
	}
	
	
	public void receive(Message msg) throws IOException {
		int ioscmd = msg.getIoscmd();
		
		if (msg.getSeqno() != seqno) {
			throw new IOException("Message out of sequence, expecting:  "+seqno+" received: "+msg.getSeqno()+" channel: "+channel.getId());
		}
		seqno++;
		if (ioscmd == MessageConstant.close) {
			out.flush();
			out.close();
		} else if (ioscmd == MessageConstant.data){
			byte[] buf = decode(msg.getData(),"UTF-8");
			counter += buf.length;	
			out.write(buf);
			out.flush();
		}
		if (Log.DEBUG) {
			Log.log(this, ""+Thread.currentThread().getId()+msg.getSeqno());
		}
	}

	private byte[] decode(String s, String charset) throws UnsupportedEncodingException {
		if (s == null || s.length() == 0) {
			return new byte[0];
		}
		byte[] buf = s.getBytes(charset);
		return Base64.decode(buf, 0, buf.length);
	}

	@Override
	public void close() throws IOException {
		super.close();
		if (out != null) {
			out.close();
			out = null;
			channel = null;
		}
	}

}
